package com.yk.spark.service.manage.app.impl;

import com.yk.spark.dao.common.api.ICommonDAO;
import com.yk.spark.model.AppModel;
import com.yk.spark.model.app.state.action.AppActionState;
import com.yk.spark.model.app.state.process.AppProcessState;
import com.yk.spark.service.manage.app.api.IAppManageService;
import com.yk.spark.service.manage.app.api.IAppService;
import com.yk.spark.service.manage.submit.api.IAppSubmitService;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLConnection;
import java.nio.charset.Charset;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * Created by Limit on 2017/8/23.
 */
@Service("appManageService")
public class AppManageServiceImpl implements IAppManageService {

    private static final String SPARK_APP_STATUS_BASE_URL="http://master.hadoop:6066/v1/submissions/status/";

    private static final String SPARK_APP_KILL_BASE_URL="http://master.hadoop:6066/v1/submissions/kill/";
    @Resource(name = "appService")
    private IAppService appService;

    @Resource(name = "appSubmitService")
    private IAppSubmitService appSubmitService;

    @Resource(name = "commonDAO")
    private ICommonDAO commonDAO;

    /**
     * 启动app
     *
     * @param appId app唯一标识
     * @return app提交状态
     */
    @Override
    public AppActionState startApp(String appId) {
        AppModel model = this.appService.getAppModel(appId);
        AppActionState appActionState = this.appSubmitService.submitSparkApp(model);

        model.setSubmissionId(appActionState.getSubmissionId());
        model.setUpdateTime(new Timestamp(new Date().getTime()));
        this.commonDAO.update(model);
        this.commonDAO.save(appActionState.getModel());
        return appActionState;
    }

    /**
     * 停止App
     *
     * @param appId app唯一标识
     * @return app停止状态
     */
    @Override
    public AppActionState stopApp(String appId) throws Exception{
        String submissionId=this.appService.getAppModel(appId).getSubmissionId();
        if(submissionId==null)
            return null;
        return new AppActionState(this.queryState(new URL(SPARK_APP_KILL_BASE_URL+submissionId)));
    }

    /**
     * 查询App执行状态
     *
     * @param appId app提交唯一标识
     * @return app执行状态
     */
    @Override
    public AppProcessState queryProcessState(String appId) throws Exception {

        String submissionId=this.appService.getAppModel(appId).getSubmissionId();
        if(submissionId==null)
            return null;
        return new AppProcessState(this.queryState(new URL(SPARK_APP_STATUS_BASE_URL+submissionId)));
    }

    private List<String> transformToList(JSONArray input){
        List<String> result=new ArrayList<>();
        for (int i = 0; i < input.length(); i++) {
            try {
                result.add(input.getString(i));
            } catch (JSONException e) {
                e.printStackTrace();
            }
        }
        return result;
    }
    private String queryState(URL url) throws JSONException {
        StringBuilder builder=new StringBuilder();
        BufferedReader bufferedReader=null;

        try {
            // 打开和URL之间的连接
            URLConnection conn = url.openConnection();
            InputStream inputStream=conn.getInputStream();
            bufferedReader = new BufferedReader(new InputStreamReader(inputStream, Charset.forName("UTF-8")));
            while (bufferedReader.ready()){
                builder.append(bufferedReader.readLine());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        finally {
            if(bufferedReader!=null)
                try {
                    bufferedReader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
        }
        return builder.toString();
    }

    /**
     * 查询App提交状态
     *
     * @param appId app提交唯一标识
     * @return app提交状态
     * @throws Exception
     */
    @Override
    public AppActionState querySubmitState(String appId) throws Exception {
        String submissionId=this.appService.getAppModel(appId).getSubmissionId();
        if(submissionId==null)
            return null;
        return new AppActionState(this.queryState(new URL(SPARK_APP_STATUS_BASE_URL+submissionId)));
    }
}
